package com.beifeng.pvcount;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;



import com.beifeng.mapreduce.MyMapReduce;


public class MyPVCountMapReduce extends Configured implements  Tool {
		//KEYIN, VALUEIN, KEYOUT, VALUEOUT 传入的Key和Value类型，传出去的Key和Value类型
		static class MyPVMap extends Mapper<LongWritable, Text, LongWritable, IntWritable> {
			LongWritable pro_Id = new LongWritable();
			IntWritable mr_value = new IntWritable(1);

			@Override
			protected void map(LongWritable key, Text value, Context context)
					throws IOException, InterruptedException {
				// TODO Auto-generated method stub
			 String line = 	value.toString();
			/* System.out.println("\n=======每一个map输入的偏移量：\n"+key.get()+"\n=============\n");*/
			 String[] words = line.split("\t");
			String province_Id_String = words[23];
			String url_String = words[1];
			//日志信息过短，有异常的日志，过滤掉
			if(words.length<=30){
				context.getCounter("我的计数器", "日志信息长度小于30").increment(1l);;
				return;
			}
			//如果url为空，直接过滤
			if(StringUtils.isBlank(url_String)){
				context.getCounter("我的计数器", "url信息为空").increment(1l);
				return;
			}
			//判断拿到的省份id是否为空
			if(StringUtils.isBlank(province_Id_String)){
				context.getCounter("我的计数器", "省份id为空").increment(1l);
				return;
			}
			//当province_Id_String不是数字的时候，也过滤掉
			int province_Id = 0;
			try{
				province_Id = Integer.parseInt(province_Id_String);
			}catch(Exception e){
				//e.printStackTrace();
				context.getCounter("我的计数器", "省份id为中文").increment(1l);
				return;
			}
			//判断省份id是否为0，为0过滤掉
			if(province_Id==0){
				context.getCounter("我的计数器", "省份id为0").increment(1l);
				return;
			}
				pro_Id.set(province_Id);
				context.write(pro_Id, mr_value);
			
			}
		}
		//KEYIN, VALUEIN, KEYOUT, VALUEOUT   输入keyvalue，输出keyvalue的类型
		static class MyPVReduce extends Reducer<LongWritable, IntWritable,LongWritable,IntWritable>{
			
			private IntWritable reduce_Value = new IntWritable();
			 
			@Override
			protected void reduce(LongWritable key, Iterable<IntWritable> values,
					Context context)
					throws IOException, InterruptedException {
				// TODO Auto-generated method stub
				System.out.println("Reduce方法开始");
				int sum = 0;
				for (IntWritable value : values) {
					sum+=value.get();
				}
				reduce_Value.set(sum);
				context.write(key, reduce_Value);
			}
		}
		
		
		public int run(String[] args) throws Exception {
			// TODO Auto-generated method stub
			//hadoop的八股文
			Configuration conf = this.getConf();
			Job job = Job.getInstance(conf, this.getClass().getSimpleName());
			//对job进行具体的配置
			
			//当你本地运行，这个设置可以不写，不会报错
			//当提价到集群上面运行的时候，这个设置不写，会报类找不到的异常 
			job.setJarByClass(MyPVCountMapReduce.class);
			
			//写一个输入路径
			Path input = new Path(args[0]);
			FileInputFormat.addInputPath(job, input);
			//写一个输出路径
			Path output = new Path(args[1]);
			FileOutputFormat.setOutputPath(job, output);
			
			//执行前先判断输出路径是否存在，存在就删除
			FileSystem fs = output.getFileSystem(conf);
			if(fs.exists(output)){
				fs.delete(output,true);
			}
			
			//设置运行的map类的相关参数
			job.setMapperClass(MyPVMap.class);
			job.setMapOutputKeyClass(LongWritable.class);
			job.setMapOutputValueClass(IntWritable.class);
			//==============shuffle=======================
			//job.setCombinerClass(MyCombiner.class);
			
			
			//==============shuffle=======================
			//设置运行的Reduce的相关参数
			job.setReducerClass(MyPVReduce.class);
			job.setOutputKeyClass(LongWritable.class);
			job.setOutputValueClass(IntWritable.class);
			
			boolean isSuccess =  job.waitForCompletion(true);
			
			
			return isSuccess?0:1;
		}
		
		public static void main(String[] args) {
			// TODO Auto-generated method stub
			Configuration conf = new Configuration();
			args = new String[]{
					"hdfs://centos6-1:8020/user/kewang/PV/datas/2015082818",
					"hdfs://centos6-1:8020/user/kewang/PV/output"
			};
		/*	MyMapReduce test =  new MyMapReduce();
			try {
				test.run(args);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}*/
			try {
				ToolRunner.run(conf, new MyPVCountMapReduce(), args);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
	 	}

	}